#[cfg(feature = "multi-threaded-cf")]
use rust_rocksdb::BoundColumnFamily;
#[cfg(not(feature = "multi-threaded-cf"))]
use rust_rocksdb::ColumnFamilyRef;

use std::fmt::Display;
use std::str::FromStr;
use std::sync::Arc;
use visit_dd_core::{Astr, FieldId, RecordId, UserId};

use crate::{CF_CACHE, CF_FV, CF_LOG, CF_META, CF_UNIQUE, ROCKSDB};

#[derive(Debug, Clone)]
pub enum VddKey {
    EntityModel(u64),
    EntityField(FieldId),
    EntityRecordMeta(RecordId),
    EntityAppUser(UserId),
    EntityActionLog(u64),
    EntityUploadFile(u64),
    EntityAppMessage(u64),
    EntityRecordChangeEvent(u64), // 记录变动事件
    IndexUser2Model(UserId),
    IndexModel2Field(u64),
    IndexModel2Record(u64),
    IndexUser2AppMessage(UserId),
    IndexUser2ActionLog(UserId), // 用户主键-日志主键
    IndexModel2ActionLog(u64),   // 模型主键-日志主键
    IndexModel2Rce(u64),         // 模型主键-记录变动事件主键
    UniquePhone(Astr),
    UniqueModelTag(Astr),
    UniqueFieldTag(Astr, Astr),      // model tag : field tag
    UniqueFieldValue(FieldId, Astr), // 字段值唯一规则
    UniqueRecordValue(u64, Astr),    // ruleID/ hash
    UniqueQueryPlan(UserId, Astr),   // 用户保存的查询
    BoundVisitRequest(u64, UserId),
    BoundVisitGroup(u64, UserId), // 模型 - 用户
    BoundRecordRule(u64, u64),
    BoundRecordAndVersion(RecordId, u64),
    BoundFieldValue(FieldId, RecordId),
    BoundModelAndSubscribe(u64, UserId), // 模型ID - 订阅者ID
    BoundRecordConsume(RecordId, u64),   // 记录ID - 模型
    BoundQueryPlan(UserId, u64),         // 用户ID - 查询计划ID
    CacheLabel(u64),
    CacheAutoIncreaseKeyId(FieldId),
}

pub(crate) trait BoundTypeKeyTrait {
    fn bound_type(&self) -> bool;
    fn get_bound_left(&self) -> u64;
    fn get_bound_right_as_u64(&self) -> Result<u64, Astr>;
}

pub(crate) trait IndexTypeKeyTrait {
    fn index_type(&self) -> bool;
    fn get_index_key_as_u64(&self) -> u64;
}

pub(crate) trait UniqueTypeKeyTrait {
    fn unique_type(&self) -> bool;
    fn get_unique_hash(&self) -> Result<Astr, Astr>;
    fn get_unique_arg(self) -> u64;
}

impl VddKey {
    fn as_u16(&self) -> u16 {
        match self {
            VddKey::EntityModel(_) => 11,
            VddKey::EntityField(_) => 12,
            VddKey::EntityRecordMeta(_) => 13,
            VddKey::EntityAppUser(_) => 14,
            VddKey::EntityActionLog(_) => 15,
            VddKey::EntityUploadFile(_) => 16,
            VddKey::EntityAppMessage(_) => 17,
            VddKey::EntityRecordChangeEvent(..) => 18,
            VddKey::IndexUser2Model(_) => 91,
            VddKey::IndexModel2Field(_) => 92,
            VddKey::IndexModel2Record(_) => 93,
            VddKey::IndexUser2AppMessage(_) => 94,
            VddKey::IndexUser2ActionLog(_) => 95,
            VddKey::IndexModel2ActionLog(_) => 96,
            VddKey::IndexModel2Rce(..) => 98,
            VddKey::UniquePhone(_) => 71,
            VddKey::UniqueModelTag(_) => 72,
            VddKey::UniqueFieldTag(..) => 73,
            VddKey::UniqueFieldValue(..) => 74,
            VddKey::UniqueRecordValue(..) => 75,
            VddKey::UniqueQueryPlan(..) => 76,
            VddKey::BoundVisitRequest(..) => 51,
            VddKey::BoundVisitGroup(..) => 52,
            VddKey::BoundRecordRule(..) => 53,
            VddKey::BoundRecordAndVersion(..) => 54,
            VddKey::BoundFieldValue(..) => 56,
            VddKey::BoundModelAndSubscribe(..) => 55,
            VddKey::BoundRecordConsume(..) => 57,
            VddKey::BoundQueryPlan(..) => 58,
            VddKey::CacheLabel(_) => 32,
            VddKey::CacheAutoIncreaseKeyId(_) => 33,
        }
    }

    pub fn entity_type(&self) -> bool {
        match self {
            VddKey::EntityModel(..)
            | VddKey::EntityField(..)
            | VddKey::EntityRecordMeta(..)
            | VddKey::EntityActionLog(..)
            | VddKey::EntityAppMessage(..)
            | VddKey::EntityAppUser(..)
            | VddKey::EntityUploadFile(..)
            | VddKey::EntityRecordChangeEvent(..) => true,
            _ => false,
        }
    }

    // 获取一个key所在的字数据库名称
    #[cfg(not(feature = "multi-threaded-cf"))]
    pub fn db_name(&self) -> Option<ColumnFamilyRef> {
        match self {
            VddKey::EntityModel(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityField(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityRecordMeta(_) => ROCKSDB.cf_handle(CF_FV),
            VddKey::EntityAppUser(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityActionLog(_) => ROCKSDB.cf_handle(CF_LOG),
            VddKey::EntityUploadFile(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityAppMessage(_) => ROCKSDB.cf_handle(CF_LOG),
            VddKey::EntityRecordChangeEvent(..) => ROCKSDB.cf_handle(CF_LOG),
            VddKey::IndexUser2Model(_) => None,
            VddKey::IndexModel2Field(_) => None,
            VddKey::IndexModel2Record(_) => None,
            VddKey::IndexUser2AppMessage(_) => None,
            VddKey::IndexUser2ActionLog(_) => None,
            VddKey::IndexModel2ActionLog(_) => None,
            VddKey::IndexModel2Rce(..) => None,
            VddKey::UniquePhone(_) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueModelTag(_) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueFieldTag(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueFieldValue(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueRecordValue(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueQueryPlan(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::BoundVisitRequest(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundVisitGroup(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundRecordRule(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundRecordAndVersion(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundFieldValue(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundModelAndSubscribe(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundRecordConsume(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundQueryPlan(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::CacheLabel(_) => ROCKSDB.cf_handle(CF_CACHE),
            VddKey::CacheAutoIncreaseKeyId(_) => ROCKSDB.cf_handle(CF_CACHE),
        }
    }

    #[cfg(feature = "multi-threaded-cf")]
    pub fn db_name(&self) -> Option<Arc<BoundColumnFamily>> {
        match self {
            VddKey::EntityModel(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityField(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityRecordMeta(_) => ROCKSDB.cf_handle(CF_FV),
            VddKey::EntityAppUser(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityActionLog(_) => ROCKSDB.cf_handle(CF_LOG),
            VddKey::EntityUploadFile(_) => ROCKSDB.cf_handle(CF_META),
            VddKey::EntityAppMessage(_) => ROCKSDB.cf_handle(CF_LOG),
            VddKey::IndexUser2Model(_) => None,
            VddKey::IndexModel2Field(_) => None,
            VddKey::IndexModel2Record(_) => None,
            VddKey::IndexUser2AppMessage(_) => None,
            VddKey::IndexUser2ActionLog(_) => None,
            VddKey::IndexModel2ActionLog(_) => None,
            VddKey::IndexSubscribe2Record(..) => None,
            VddKey::UniquePhone(_) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueModelTag(_) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueFieldTag(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueFieldValue(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::UniqueRecordValue(..) => ROCKSDB.cf_handle(CF_UNIQUE),
            VddKey::BoundVisitRequest(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundVisitGroup(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundRecordRule(..) => ROCKSDB.cf_handle(CF_META),
            VddKey::BoundRecordAndVersion(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundFieldValue(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundModelAndSubscribe(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::BoundRecordConsume(..) => ROCKSDB.cf_handle(CF_FV),
            VddKey::CacheParquetRefreshFlag(_) => ROCKSDB.cf_handle(CF_CACHE),
            VddKey::CacheLabel(_) => ROCKSDB.cf_handle(CF_CACHE),
            VddKey::CacheAutoIncreaseKeyId(_) => ROCKSDB.cf_handle(CF_CACHE),
        }
    }
}

impl TryFrom<&str> for VddKey {
    type Error = Astr;

    fn try_from(value: &str) -> Result<Self, Self::Error> {
        fn _to_u64(source: Option<&Astr>) -> u64 {
            source
                .map(|astr| u64::from_str(&astr).unwrap_or(0))
                .unwrap_or(0)
        }
        let k_str = if value.starts_with("/") {
            &value[1..]
        } else {
            value
        };
        let ks = k_str
            .split("/")
            .map(|item| Arc::from(item))
            .collect::<Vec<Astr>>();
        let code = ks.first().unwrap().parse::<u64>().unwrap();
        if ks.len() < 2 {
            Err("invalid key type".into())
        } else {
            let key = match code {
                11 => Some(VddKey::EntityModel(_to_u64(ks.get(1)))),
                12 => Some(VddKey::EntityField(FieldId(_to_u64(ks.get(1))))),
                13 => Some(VddKey::EntityRecordMeta(RecordId(_to_u64(ks.get(1))))),
                14 => Some(VddKey::EntityAppUser(UserId(_to_u64(ks.get(1))))),
                15 => Some(VddKey::EntityActionLog(_to_u64(ks.get(1)))),
                16 => Some(VddKey::EntityUploadFile(_to_u64(ks.get(1)))),
                17 => Some(VddKey::EntityAppMessage(_to_u64(ks.get(1)))),
                18 => Some(VddKey::EntityRecordChangeEvent(_to_u64(ks.get(1)))),
                91 => Some(VddKey::IndexUser2Model(UserId(_to_u64(ks.get(1))))),
                92 => Some(VddKey::IndexModel2Field(_to_u64(ks.get(1)))),
                93 => Some(VddKey::IndexModel2Record(_to_u64(ks.get(1)))),
                94 => Some(VddKey::IndexUser2AppMessage(UserId(_to_u64(ks.get(1))))),
                95 => Some(VddKey::IndexUser2ActionLog(UserId(_to_u64(ks.get(1))))),
                96 => Some(VddKey::IndexModel2ActionLog(_to_u64(ks.get(1)))),
                98 => Some(VddKey::IndexModel2Rce(_to_u64(ks.get(1)))),
                71 => Some(VddKey::UniquePhone(ks.get(1).unwrap().clone())),
                72 => Some(VddKey::UniqueModelTag(ks.get(1).unwrap().clone())),
                73 => ks
                    .get(1)
                    .map(|tags| {
                        let tag_ary = tags
                            .split(":")
                            .map(|tag| Arc::from(tag))
                            .collect::<Vec<Astr>>();
                        Some(VddKey::UniqueFieldTag(
                            tag_ary.get(0).unwrap().clone(),
                            tag_ary.get(1).unwrap().clone(),
                        ))
                    })
                    .unwrap_or(None),
                74 => Some(VddKey::UniqueFieldValue(
                    FieldId(_to_u64(ks.get(1))),
                    ks.get(2).unwrap_or(&"".into()).clone(),
                )),
                75 => Some(VddKey::UniqueRecordValue(
                    _to_u64(ks.get(1)),
                    ks.get(2).unwrap_or(&"".into()).clone(),
                )),
                76 => Some(VddKey::UniqueQueryPlan(
                    UserId(_to_u64(ks.get(1))),
                    ks.get(2).unwrap_or(&"".into()).clone(),
                )),
                51 => Some(VddKey::BoundVisitRequest(
                    _to_u64(ks.get(1)),
                    UserId(_to_u64(ks.get(2))),
                )),
                52 => Some(VddKey::BoundVisitGroup(
                    _to_u64(ks.get(1)),
                    UserId(_to_u64(ks.get(2))),
                )),
                53 => Some(VddKey::BoundRecordRule(
                    _to_u64(ks.get(1)),
                    _to_u64(ks.get(2)),
                )),
                54 => Some(VddKey::BoundRecordAndVersion(
                    RecordId(_to_u64(ks.get(1))),
                    _to_u64(ks.get(2)),
                )),
                55 => Some(VddKey::BoundModelAndSubscribe(
                    _to_u64(ks.get(1)),
                    UserId(_to_u64(ks.get(2))),
                )),
                56 => Some(VddKey::BoundFieldValue(
                    FieldId(_to_u64(ks.get(1))),
                    RecordId(_to_u64(ks.get(2))),
                )),
                57 => Some(VddKey::BoundRecordConsume(
                    RecordId(_to_u64(ks.get(1))),
                    _to_u64(ks.get(2)),
                )),
                58 => Some(VddKey::BoundQueryPlan(
                    UserId(_to_u64(ks.get(1))),
                    _to_u64(ks.get(2)),
                )),
                32 => Some(VddKey::CacheLabel(_to_u64(ks.get(1)))),
                33 => Some(VddKey::CacheAutoIncreaseKeyId(FieldId(_to_u64(ks.get(1))))),
                _ => None,
            };
            match key {
                Some(key) => Ok(key),
                None => Err("invalid key type".into()),
            }
        }
    }
}

impl From<&VddKey> for u16 {
    fn from(value: &VddKey) -> Self {
        value.as_u16()
    }
}

impl From<VddKey> for u16 {
    fn from(value: VddKey) -> Self {
        value.as_u16()
    }
}

impl Display for VddKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut s = String::new();
        let c: u16 = self.into();
        s.push_str(&format!("/{}/", c));
        match self {
            VddKey::EntityModel(m)
            | VddKey::IndexModel2Field(m)
            | VddKey::IndexModel2Record(m)
            | VddKey::IndexModel2ActionLog(m)
            | VddKey::IndexModel2Rce(m) => {
                if *m > 0 {
                    s.push_str(&format!("{}", m));
                }
            }
            VddKey::EntityField(f) | VddKey::CacheAutoIncreaseKeyId(f) => {
                if f.0 > 0 {
                    s.push_str(&format!("{}", f.0));
                }
            }
            VddKey::EntityRecordMeta(r) => {
                if r.0 > 0 {
                    s.push_str(&format!("{}", r.0));
                }
            }
            VddKey::EntityAppUser(u)
            | VddKey::IndexUser2Model(u)
            | VddKey::IndexUser2AppMessage(u)
            | VddKey::IndexUser2ActionLog(u) => {
                if u.0 > 0 {
                    s.push_str(&format!("{}", u.0));
                }
            }
            VddKey::EntityActionLog(id)
            | VddKey::EntityUploadFile(id)
            | VddKey::EntityAppMessage(id)
            | VddKey::CacheLabel(id)
            | VddKey::EntityRecordChangeEvent(id) => {
                if *id > 0 {
                    s.push_str(&format!("{}", id));
                }
            }
            VddKey::UniquePhone(hash) | VddKey::UniqueModelTag(hash) => {
                if !hash.is_empty() {
                    s.push_str(&format!("{}", hash));
                }
            }
            VddKey::UniqueFieldTag(m_tag, f_tag) => {
                if !m_tag.is_empty() {
                    s.push_str(&format!("{}:", m_tag));
                    if !f_tag.is_empty() {
                        s.push_str(&format!("{}", f_tag));
                    }
                }
            }
            VddKey::UniqueFieldValue(f, hash) => {
                s.push_str(&format!("{}/", f.0));
                if !hash.is_empty() {
                    s.push_str(&format!("{}", hash));
                }
            }
            VddKey::UniqueRecordValue(id, hash) => {
                if *id > 0 {
                    s.push_str(&format!("{}/", id));
                    if !hash.is_empty() {
                        s.push_str(&format!("{}", hash));
                    }
                }
            }
            VddKey::UniqueQueryPlan(u, hash) => {
                if u.0 > 0 {
                    s.push_str(&format!("{}/", u.0));
                    if !hash.is_empty() {
                        s.push_str(&format!("{}", hash));
                    }
                }
            }
            VddKey::BoundRecordRule(m, id) => {
                s.push_str(&format!("{}/", m));
                if *id > 0 {
                    s.push_str(&format!("{}", id));
                }
            }
            VddKey::BoundRecordAndVersion(r, id) => {
                s.push_str(&format!("{}/", r.0));
                if *id > 0 {
                    s.push_str(&format!("{}", id));
                }
            }
            VddKey::BoundFieldValue(f, r) => {
                s.push_str(&format!("{}/", f.0));
                if r.0 > 0 {
                    s.push_str(&format!("{}", r.0));
                }
            }
            VddKey::BoundRecordConsume(r, m_c) => {
                if r.0 > 0 {
                    s.push_str(&format!("{}/", r.0));
                    if *m_c > 0 {
                        s.push_str(&format!("{}", m_c));
                    }
                }
            }
            VddKey::BoundVisitRequest(m, u)
            | VddKey::BoundVisitGroup(m, u)
            | VddKey::BoundModelAndSubscribe(m, u) => {
                s.push_str(&format!("{}/", m));
                if u.0 > 0 {
                    s.push_str(&format!("{}", u.0));
                }
            }
            VddKey::BoundQueryPlan(u, id) => {
                if u.0 > 0 {
                    s.push_str(&format!("{}/", u.0));
                }
                if *id > 0 {
                    s.push_str(&format!("{}", id));
                }
            }
        }
        write!(f, "{}", s)
    }
}

impl BoundTypeKeyTrait for VddKey {
    fn bound_type(&self) -> bool {
        match self {
            VddKey::BoundVisitRequest(..)
            | VddKey::BoundVisitGroup(..)
            | VddKey::BoundFieldValue(..)
            | VddKey::BoundModelAndSubscribe(..)
            | VddKey::BoundRecordRule(..)
            | VddKey::BoundRecordAndVersion(..)
            | VddKey::BoundRecordConsume(..)
            | VddKey::BoundQueryPlan(..) => true,
            _ => false,
        }
    }
    fn get_bound_left(&self) -> u64 {
        match self {
            VddKey::BoundVisitRequest(m, ..) | VddKey::BoundVisitGroup(m, ..) => *m,
            VddKey::BoundFieldValue(f, ..) => f.0,
            VddKey::BoundModelAndSubscribe(m, ..) => *m,
            VddKey::BoundRecordRule(m, ..) => *m,
            VddKey::BoundRecordAndVersion(r, ..) => r.0,
            VddKey::BoundRecordConsume(r, ..) => r.0,
            VddKey::BoundQueryPlan(u, ..) => u.0,
            _ => {
                panic!("Not bound type")
            }
        }
    }

    fn get_bound_right_as_u64(&self) -> Result<u64, Astr> {
        match self {
            VddKey::BoundVisitRequest(_m, u)
            | VddKey::BoundVisitGroup(_m, u)
            | VddKey::BoundModelAndSubscribe(_m, u) => Ok(u.0),
            VddKey::BoundFieldValue(_f, r) => Ok(r.0),
            VddKey::BoundRecordRule(_m, id) => Ok(*id),
            VddKey::BoundRecordAndVersion(_r, id) => Ok(*id),
            VddKey::BoundRecordConsume(_r, id) => Ok(*id),
            VddKey::BoundQueryPlan(_u, id) => Ok(*id),
            _ => Err("Not bound type".into()),
        }
    }
}

impl AsRef<VddKey> for VddKey {
    fn as_ref(&self) -> &VddKey {
        self
    }
}

impl IndexTypeKeyTrait for VddKey {
    fn index_type(&self) -> bool {
        match self {
            VddKey::IndexUser2Model(..)
            | VddKey::IndexModel2Field(..)
            | VddKey::IndexModel2Record(..)
            | VddKey::IndexUser2AppMessage(..)
            | VddKey::IndexUser2ActionLog(..)
            | VddKey::IndexModel2ActionLog(..)
            | VddKey::IndexModel2Rce(..) => true,
            _ => false,
        }
    }
    fn get_index_key_as_u64(&self) -> u64 {
        match self {
            VddKey::IndexModel2Field(m)
            | VddKey::IndexModel2Record(m)
            | VddKey::IndexModel2ActionLog(m)
            | VddKey::CacheLabel(m)
            | VddKey::IndexModel2Rce(m) => *m,
            VddKey::CacheAutoIncreaseKeyId(f) => f.0,
            VddKey::IndexUser2ActionLog(u)
            | VddKey::IndexUser2AppMessage(u)
            | VddKey::IndexUser2Model(u) => u.0,
            _ => {
                panic!("Not index type")
            }
        }
    }
}

impl UniqueTypeKeyTrait for VddKey {
    fn unique_type(&self) -> bool {
        match self {
            VddKey::UniquePhone(_)
            | VddKey::UniqueModelTag(_)
            | VddKey::UniqueFieldTag(..)
            | VddKey::UniqueFieldValue(..)
            | VddKey::UniqueQueryPlan(..)
            | VddKey::UniqueRecordValue(..) => true,
            _ => false,
        }
    }

    fn get_unique_hash(&self) -> Result<Astr, Astr> {
        match self {
            VddKey::UniquePhone(h) => Ok(h.clone()),
            VddKey::UniqueModelTag(h) => Ok(h.clone()),
            VddKey::UniqueFieldTag(h1, h2) => Ok(format!("{}:{}", h1, h2).into()),
            VddKey::UniqueFieldValue(_f, h) => Ok(h.clone()),
            VddKey::UniqueRecordValue(_rule_id, h) => Ok(h.clone()),
            VddKey::UniqueQueryPlan(_u, h) => Ok(h.clone()),
            _ => Err("Not unique type".into()),
        }
    }

    fn get_unique_arg(self) -> u64 {
        match self {
            VddKey::UniqueFieldValue(f, _) => f.0,
            VddKey::UniqueRecordValue(rule_id, _) => rule_id,
            VddKey::UniqueQueryPlan(u, _) => u.0,
            _ => 0,
        }
    }
}